package com.runssnail.demo.kafka;

import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.KafkaConsumer;

/**
 * KafkaConsumer
 * 
 * @author zhengwei
 *
 */
public class ConsumerNewApiTest {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // kafka server
//        props.put("zookeeper.connect", "localhost:2181");

        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // key Deserializer
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // value Deserializer

        props.put("partition.assignment.strategy", "range");

        props.put("group.id", "test1"); // group
        
        props.put("session.timeout.ms", "1000");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "10000");
        
        KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>(props);
        
        consumer.subscribe("topic-test"); // topic
        
        boolean isRunning = true;
        while (isRunning) {
            Map records = consumer.poll(200);
            // process(records);
            if (records != null && !records.isEmpty()) {
                System.out.println(records);
            }
        }
        // consumer.close();

    }

}
